package com.dc.dbus.session.register;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import com.dc.dbus.session.Log;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import com.dc.dbus.transport.commons.ServerAddress;
import com.dc.dbus.transport.commons.ServiceChangeListener;

public class ZookeeperServiceRegister implements ServiceRegister {

	protected final String zkHostPort;
	protected ZooKeeper zk;
	
	protected final ServerAddress serverAddress;
	
	protected ConcurrentHashMap<String, ServiceInfo> serviceNameMapInfo = new ConcurrentHashMap<>();
	
	protected ConcurrentHashMap<String, String> listenerServiceNames = new ConcurrentHashMap<>();
	
	protected final ServiceChangeListener listener;
	
	protected final static ServiceChangeListener NULL_LISTENER = new ServiceChangeListener() {
		@Override
		public void serviceActive(String serviceName, String node, String[] routers, String[] hosts, int[] ports) {
		}
		@Override
		public void serviceDisconected(String serviceName, String node) {
		}
		@Override
		public void serviceChange(String serviceName, String node, String[] routers, String[] hosts, int[] ports) {
		}
	};

	protected final Watcher watcher = new Watcher() {

		@Override
		public void process(WatchedEvent event) {

			synchronized (ZookeeperServiceRegister.this) {

				EventType eventType = event.getType();
				KeeperState keeperState = event.getState();
				String watchPath = event.getPath();

				Log.registerLogger.info("WatchedEvent: keeperState:" + keeperState + " eventType:" + eventType + " watchPath:" + watchPath);

				switch (keeperState) {

					case Expired:

						try {

							Log.registerLogger.warn("zk Expired try reConnected !");
							// 如果恢复连接，但超过了sessionTimeout，依旧尝试恢复

							reRegisterThreadInterrupt = true;

							try {
								reRegisterThread.join();
							} catch (Exception e2) {
								Log.registerLogger.error("", e2);
							}
							reRegisterThreadInterrupt = false;
							reRegisterThread = new Thread(reRegister);

							reRegisterThread.start();

						} catch (Exception e) {
							Log.registerLogger.error("", e);
						}

						break;

					default:

						switch (eventType) {
							case NodeChildrenChanged:
								handleNodeChanged(watchPath, eventType);
								break;

							case NodeDataChanged:
								handleNodeDataChanged(watchPath, eventType);
								break;

							case NodeDeleted:
								handleNodeChanged(watchPath.substring(0, watchPath.lastIndexOf("/")), eventType);
								break;

							case NodeCreated:
							default:

								if (watchPath != null && watchPath.length() > 0) {

									try {
										zk.getChildren(watchPath, true);
									} catch (Exception e) {
										Log.registerLogger.error("", e);
									}
									try {
										zk.getData(watchPath, true, null);
									} catch (Exception e) {
										Log.registerLogger.error("", e);
									}
								}
								break;
						}

						break;
				}
			}
		}
	};

	protected volatile boolean reRegisterThreadInterrupt = false;
	
	protected final Runnable reRegister = new Runnable() {
		
		@Override
		public void run() {
			
			// 有任何异常则重新执行，直到成功
			for (;;) {
				try {
					
					if (reRegisterThreadInterrupt) {
						break;
					}
					
					try {
						zk.close();
					} catch (Exception e1) {
						e1.printStackTrace();
					}

					if (reRegisterThreadInterrupt) {
						break;
					}

					linkZookeeper(ZookeeperServiceRegister.this.zkHostPort, 
							ZookeeperServiceRegister.this.sessionTimeout);
					
					if (reRegisterThreadInterrupt) {
						break;
					}
					
					for (ServiceInfo service : serviceNameMapInfo.values()) {
						_registerService(service.serviceName, service.routers, service.serviceHosts, service.servicePorts);
					}
					
					if (reRegisterThreadInterrupt) {
						break;
					}
					
					for (String serviceName : listenerServiceNames.values()) {
						handleNodeChanged("/" + serviceName, null);
//						_registerServiceListener(serviceName);
					}
					
					break;
				} catch (Exception e) {

					Log.registerLogger.error("", e);
					
					try {
						TimeUnit.MILLISECONDS.sleep(500);
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
				}
			}
			
		}
	};
	
	protected Thread reRegisterThread = new Thread(reRegister);
	
	protected final int sessionTimeout;
	
	public static ZookeeperServiceRegister getServerSideReiger(String zkHostPort,
			ServerAddress serverAddress) 
			throws IOException {
		
		return getServerSideReiger(zkHostPort, 500, serverAddress);
	}
	
	public static ZookeeperServiceRegister getClientSideReiger(String zkHostPort,
			ServiceChangeListener listener) 
			throws IOException {
		
		return getClientSideReiger(zkHostPort, 500, listener);
	}
	
	public static ZookeeperServiceRegister getServerSideReiger(String zkHostPort, int sessionTimeout, 
			ServerAddress serverAddress) 
			throws IOException {
		
		return new ZookeeperServiceRegister(zkHostPort, sessionTimeout, serverAddress, null);
	}
	
	public static ZookeeperServiceRegister getClientSideReiger(String zkHostPort, int sessionTimeout, 
			ServiceChangeListener listener) 
			throws IOException {
		
		return new ZookeeperServiceRegister(zkHostPort, sessionTimeout, null, listener);
	}
	
	public ZookeeperServiceRegister(String zkHostPort, int sessionTimeout, 
			ServerAddress serverAddress, ServiceChangeListener listener) 
			throws IOException {
		
		this.zkHostPort = zkHostPort;
		this.sessionTimeout = sessionTimeout;
		
		this.serverAddress = serverAddress;
		
		this.listener = listener == null ? NULL_LISTENER : listener;
		
		linkZookeeper(zkHostPort, sessionTimeout);
	}
	
	protected synchronized void linkZookeeper(String zkHostPort, int sessionTimeout) throws IOException {

		ZooKeeper zk = new ZooKeeper(zkHostPort, sessionTimeout, watcher);
		this.zk = zk;
	}
	
	class ServiceInfo {
		
		String serviceName;
		
		String[] routers;
		String[] serviceHosts;
		int[] servicePorts;
		
		String nodePath;
		
		public ServiceInfo(String serviceName, String[] routers, String[] serviceHosts, int[] servicePorts) {
			super();
			this.serviceName = serviceName;
			this.routers = routers;
			this.serviceHosts = serviceHosts;
			this.servicePorts = servicePorts;
		}
	}
	
	@Override
	public synchronized String registerService(String serviceName, String[] routers) {

		if (serviceNameMapInfo.contains(serviceName)) {
			throw new RuntimeException(serviceName + " has been registed");
		} else {
			
			String[] serviceHosts = serverAddress.getIps();
			int[] servicePorts = serverAddress.getPorts();
			
			ServiceInfo serviceInfo = new ServiceInfo(serviceName, routers, serviceHosts, servicePorts);
			String nodePath = _registerService(serviceName, routers, serviceHosts, servicePorts);
			serviceInfo.nodePath = nodePath;
			serviceNameMapInfo.put(serviceName, serviceInfo);

			return nodePath;
		}
	}
	
	@Override
	public synchronized void updateServiceRouter(String serviceName, String[] newRouters) {
		
		try {
			
			ServiceInfo serviceInfo = serviceNameMapInfo.get(serviceName);
			if (serviceInfo == null) {
				throw new RuntimeException("should invoke method \"registerService\" before \"updateServiceRouter\"");
			}

			Log.registerLogger.info("updateServiceRouter BEGIN: serviceName: " + serviceName + ", path:" + serviceInfo.nodePath + ", newRouters:" + Arrays.toString(newRouters) );
			
			createServiceBasePath(serviceName);
			
			byte[] recivedata = zk.getData(serviceInfo.nodePath, true, null);
			String data = new String(recivedata, "UTF-8");
			Log.registerLogger.info("updateServiceRouter: get old routers info from path:" + serviceInfo.nodePath + ":data:" + data);
			
			StringBuilder serviceId = new StringBuilder();
			for (int i=0; ; ) {
				serviceId.append(newRouters[i]);
				
				i++;
				if (i == newRouters.length) {
					break;
				} else {
					serviceId.append(",");
				}
			}
			
			String newData = serviceId.append(data.substring(data.indexOf(';'), data.length())).toString();
			
			zk.setData(serviceInfo.nodePath, newData.getBytes(), -1);

			Log.registerLogger.info("updateServiceRouter SUCCESS: serviceName: " + serviceName + ", path:" + serviceInfo.nodePath + ", data:" + newData);
			
			recivedata = zk.getData(serviceInfo.nodePath, true, null);
			data = new String(recivedata, "UTF-8");
			Log.registerLogger.info("updateServiceRouter: get new routers info from path:" + serviceInfo.nodePath + ":data:" + data);
			
		} catch (KeeperException e) {
			
			if (e.code() == KeeperException.Code.CONNECTIONLOSS) {
				throw new RuntimeException("can't link to zookeeper", e);
			} else {
				throw new RuntimeException(e);
			}
			
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		
	}
	
	protected synchronized String _registerService(String serviceName, String[] routers, String[] serviceHosts, int[] servicePorts) {
		
		try {

			Log.registerLogger.info("registerService BEGIN: registerService:" + serviceName);

			createServiceBasePath(serviceName);
			
			StringBuilder serviceId = new StringBuilder();
			for (int i=0; ; ) {
				serviceId.append(routers[i]);
				
				i++;
				if (i == routers.length) {
					break;
				} else {
					serviceId.append(",");
				}
			}
			
			serviceId.append(";");
			
			for (int i=0; ; ) {
				serviceId.append(serviceHosts[i]).append(":").append(servicePorts[i]);
				
				i++;
				if (i == serviceHosts.length) {
					break;
				} else {
					serviceId.append(",");
				}
			}
			
			String nodePath = zk.create("/" + serviceName + "/node_", 
					serviceId.toString().getBytes(), 
					ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

			Log.registerLogger.info("registerService SUCCESS: register zk node: " + nodePath);
			
			return nodePath;
			
		} catch (KeeperException e) {
			
			if (e.code() == KeeperException.Code.CONNECTIONLOSS) {
				throw new RuntimeException("can't link to zookeeper", e);
			} else {
				throw new RuntimeException(e);
			}
			
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		
	}
	
	protected void createServiceBasePath(String serviceName) {
		
		try {
			zk.create("/" + serviceName, serviceName.getBytes(), 
					ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		} catch (KeeperException e) {
			
			if (e.code() == KeeperException.Code.NODEEXISTS) {
				return;
			} else if (e.code() == KeeperException.Code.CONNECTIONLOSS) {
				throw new RuntimeException("can't link to zookeeper", e);
			} else {
				throw new RuntimeException(e);
			}
			
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}
	
	protected HashMap<String, HashMap<String, String>> serviceNameMapNodeNameMapData = new HashMap<>();
	
	protected void putNodeDataToCache(String serviceName, String nodeName, String data) {
		
		HashMap<String, String> nodeNameMapData = serviceNameMapNodeNameMapData.get(serviceName);
		if (nodeNameMapData == null) {
			nodeNameMapData = new HashMap<>();
			serviceNameMapNodeNameMapData.put(serviceName, nodeNameMapData);
		}
		
		nodeNameMapData.put(nodeName, data);
	}
	
	protected String getNodeDataFromCache(String serviceName, String nodeName) {
		
		HashMap<String, String> nodeNameMapData = serviceNameMapNodeNameMapData.get(serviceName);
		if (nodeNameMapData == null) {
			return null;
		}
		
		return nodeNameMapData.get(nodeName);
	}
	
	protected String removeNodeDataFromCache(String serviceName, String nodeName) {
		
		HashMap<String, String> nodeNameMapData = serviceNameMapNodeNameMapData.get(serviceName);
		if (nodeNameMapData == null) {
			return null;
		}
		
		return nodeNameMapData.remove(nodeName);
	}
	
	@Override
	public synchronized void registerServiceListener(String serviceName)  {
		
		if (!listenerServiceNames.contains(serviceName)) {
			
			_registerServiceListener(serviceName);
			
			listenerServiceNames.put(serviceName, serviceName);
		}
	}
	
	protected synchronized void _registerServiceListener(String serviceName) {

		Log.registerLogger.info("registerServiceListener BEGIN: registerServiceListener:" + serviceName);

		createServiceBasePath(serviceName);
		
		String watchPath = "/" + serviceName;
		
		try {
			List<String> childrenPaths = zk.getChildren(watchPath, true);
			
			for (String nodeName : childrenPaths) {
				
				try {
					
					String childPath = watchPath + "/" + nodeName;
					
					byte[] recivedata = zk.getData(childPath, true, null);
					String data = new String(recivedata, "UTF-8");
					Log.registerLogger.info("registerServiceListener: receive the path:" + childPath + " data:" + data);
					
					putNodeDataToCache(serviceName, nodeName, data);

					int routerEndIndex = data.indexOf(';');
					String routerInfo = data.substring(0, routerEndIndex);
					String[] routers = routerInfo.split(",");
					
					String ipAndPorts = data.substring(routerEndIndex+1, data.length());
					String[] hostAndPorts = ipAndPorts.split(",");
					String[] hosts = new String[hostAndPorts.length];
					int[] ports = new int[hostAndPorts.length];

					for (int i=0; i<hostAndPorts.length; i++) {
						String hostAndPort = hostAndPorts[i];
						
						String[] hostPort = hostAndPort.split(":");
						hosts[i] = hostPort[0];
						ports[i] = Integer.parseInt(hostPort[1]);
					}
					
					listener.serviceActive(serviceName, nodeName, routers, hosts, ports);
					
				} catch (Exception e) {
					Log.registerLogger.error("", e);
				}
				
		    }

			Log.registerLogger.info("registerServiceListener SUCCESS: registerServiceListener:" + serviceName);
			
		} catch (Exception e1) {
			throw new RuntimeException(e1);
		}
	}
	
	protected synchronized void handleNodeChanged(String path, EventType eventType) {
		
		try {

			Log.registerLogger.info("handleNodeChanged BEGIN: path:" + path);

			List<String> childrenPaths = zk.getChildren(path, true);
			
			String serviceName = path.substring(1);
			
			HashMap<String, String> nodeMapDataCache = serviceNameMapNodeNameMapData.get(serviceName);
			HashMap<String, String> oldNodePathMapData = null;
			if (nodeMapDataCache != null) {
				oldNodePathMapData = new HashMap<>(nodeMapDataCache);
			} else {
				oldNodePathMapData = new HashMap<>();
			}
			
			for (String nodeName : childrenPaths) {
				
				try {
					
					String childPath = path + "/" + nodeName;
					
					byte[] recivedata = zk.getData(childPath, true, null);
					String data = new String(recivedata, "UTF-8");
					Log.registerLogger.info("handleNodeChanged: receive the path:" + childPath + ":data:" + data);
					
					int routerEndIndex = data.indexOf(';');
					String routerInfo = data.substring(0, routerEndIndex);
					String[] routers = routerInfo.split(",");
					
					String ipAndPorts = data.substring(routerEndIndex+1, data.length());
					String[] hostAndPorts = ipAndPorts.split(",");
					String[] hosts = new String[hostAndPorts.length];
					int[] ports = new int[hostAndPorts.length];

					for (int i=0; i<hostAndPorts.length; i++) {
						String hostAndPort = hostAndPorts[i];
						
						String[] hostPort = hostAndPort.split(":");
						hosts[i] = hostPort[0];
						ports[i] = Integer.parseInt(hostPort[1]);
					}
					
					String oldData = oldNodePathMapData.remove(nodeName);
					if (oldData == null) {
						
						listener.serviceActive(serviceName, nodeName, routers, hosts, ports);
						putNodeDataToCache(serviceName, nodeName, data);
						
					} else {
						
						if (!oldData.equals(data)) {
							
							listener.serviceChange(serviceName, nodeName, routers, hosts, ports);
							
							putNodeDataToCache(serviceName, nodeName, data);
						}
					}
					
				} catch (Exception e) {
					Log.registerLogger.error("", e);
				}
				
		    }
			
			for (String nodeName : oldNodePathMapData.keySet()) {
				
				listener.serviceDisconected(serviceName, nodeName);
				
				removeNodeDataFromCache(serviceName, nodeName);
			}

			Log.registerLogger.info("handleNodeChanged SUCCESS: path:" + path);
			
		} catch (Exception e1) {
			Log.registerLogger.error("", e1);
		}
	}
	
	protected synchronized void handleNodeDataChanged(String childPath, EventType eventType) {
		
		try {

			Log.registerLogger.info("handleNodeDataChanged BEGIN: childPath:" + childPath);
			
			int serviceNameEndIndex = childPath.indexOf('/', 1);
			String serviceName = childPath.substring(1, serviceNameEndIndex);
			String nodeName = childPath.substring(serviceNameEndIndex+1, childPath.length());
			
			HashMap<String, String> nodeMapDataCache = serviceNameMapNodeNameMapData.get(serviceName);
			HashMap<String, String> oldNodePathMapData = null;
			if (nodeMapDataCache != null) {
				oldNodePathMapData = new HashMap<>(nodeMapDataCache);
			} else {
				oldNodePathMapData = new HashMap<>();
			}
			
			byte[] recivedata = zk.getData(childPath, true, null);
			String data = new String(recivedata, "UTF-8");
			Log.registerLogger.info("handleNodeDataChanged: receive the path:" + childPath + ":data:" + data);
			
			int routerEndIndex = data.indexOf(';');
			String routerInfo = data.substring(0, routerEndIndex);
			String[] routers = routerInfo.split(",");
			
			String ipAndPorts = data.substring(routerEndIndex+1, data.length());
			String[] hostAndPorts = ipAndPorts.split(",");
			String[] hosts = new String[hostAndPorts.length];
			int[] ports = new int[hostAndPorts.length];

			for (int i=0; i<hostAndPorts.length; i++) {
				String hostAndPort = hostAndPorts[i];
				
				String[] hostPort = hostAndPort.split(":");
				hosts[i] = hostPort[0];
				ports[i] = Integer.parseInt(hostPort[1]);
			}
			
			String oldData = oldNodePathMapData.remove(nodeName);
			if (oldData == null) {
				
				listener.serviceActive(serviceName, nodeName, routers, hosts, ports);
				putNodeDataToCache(serviceName, nodeName, data);
				
			} else {
				
				if (!oldData.equals(data)) {
					
					listener.serviceChange(serviceName, nodeName, routers, hosts, ports);
					
					putNodeDataToCache(serviceName, nodeName, data);
				}
			}

			Log.registerLogger.info("handleNodeDataChanged SUCCESS: childPath:" + childPath);
			
		} catch (Exception e1) {
			Log.registerLogger.error("", e1);
		}
	}
	
	public void close() throws InterruptedException {
		
		zk.close();
	}
	
}
